#ifndef BROKER_H
#define BROKER_H

#include <google/protobuf/message.h>
#include <muduo/proto/codec.h>
#include <muduo/proto/dispatcher.h>
#include <muduo/base/Logging.h>
#include <muduo/net/EventLoop.h>
#include <muduo/net/TcpServer.h>
#include "../common/project.pb.h"
#include "../common/threadpool.hpp"
#include "../common/message.pb.h"
#include "../common/Log.hpp"
#include "Channel.hpp"
#include "Connection.hpp"
#include "Consumer.hpp"
#include "Host.hpp"
using namespace std;

const string dbfile = "/meta.db";
const string hostname = "MyVirtualHost";

class Server 
{
public:
    typedef shared_ptr<google::protobuf::Message> MessagePtr;
    /*

    把服务器进行初始化，传递一个端口号和根目录，要初始化的内容有：
    1. 服务器 2. 分发器 3. Protubuf协议解析器 4. 虚拟机 5. 消费者 6. 连接管理 7. 线程池

    1. 服务器
    muduo库中的接口：对于服务器来说，要指定EventLoop，ip端口号，名字和选项
    TcpServer(EventLoop* loop, const InetAddress& listenAddr, const string& nameArg, Option option = kNoReusePort);

    2. 分发器：遇到不同情况怎么处理
    muduo库中的接口：需要指定一个defaultCb
    ProtobufDispatcher(const ProtobufMessageCallback& defaultCb);

    3. Protubuf协议解析器：进行Protubuf的协议解析
    muduo库中的接口：需要一个ProtobufMessageCallback函数回调
    ProtobufCodec(const ProtobufMessageCallback& messageCb);

    4. 虚拟机：前面自定义类型
    VirtualHost(const string &hname, const string &basedir, const string &dbfile);
    5. 消费者：前面自定义类型
    6. 连接：前面自定义类型
    7. 线程池：前面自定义类型
    */
    
    Server(int port, const string &basedir)
        : _server(&_baseloop, muduo::net::InetAddress("0.0.0.0", port), 
        "Server", muduo::net::TcpServer::kReusePort),
        _dispatcher(bind(&Server::onUnknownMessage, this, placeholders::_1, 
            placeholders::_2, placeholders::_3)),
        _codec(make_shared<ProtobufCodec>(bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher, 
            placeholders::_1, placeholders::_2, placeholders::_3))),
        _virtual_host(make_shared<VirtualHost>(hostname, basedir, basedir + dbfile)),
        _consumer_manager(make_shared<ConsumerManager>()),
        _connection_manager(make_shared<ConnectionManager>()),
        _threadpool(make_shared<threadpool>())
    {
        // 把历史消息中的队列加载进来
        QueueMap qm = _virtual_host->Allqueues();
        for (auto &q : qm)
            _consumer_manager->InitQueueConsumer(q.first);
        // 给事件分发器进行注册业务请求处理函数
        // 打开信道
        _dispatcher.registerMessageCallback<MessageQueue::openChannelRequest>(bind(&Server::onOpenChannel, this, 
            placeholders::_1, placeholders::_2, placeholders::_3));
        // 关闭信道
        _dispatcher.registerMessageCallback<MessageQueue::closeChannelRequest>(bind(&Server::onCloseChannel, this, 
            placeholders::_1, placeholders::_2, placeholders::_3));
        // 声明交换机
        _dispatcher.registerMessageCallback<MessageQueue::declareExchangeRequest>(bind(&Server::onDeclareExchange, this, 
            placeholders::_1, placeholders::_2, placeholders::_3));
        // 删除交换机
        _dispatcher.registerMessageCallback<MessageQueue::deleteExchangeRequest>(bind(&Server::onDeleteExchange, this, 
            placeholders::_1, placeholders::_2, placeholders::_3));
        // 声明队列
        _dispatcher.registerMessageCallback<MessageQueue::declareQueueRequest>(bind(&Server::onDeclareQueue, this, 
            placeholders::_1, placeholders::_2, placeholders::_3));
        // 删除队列
        _dispatcher.registerMessageCallback<MessageQueue::deleteQueueRequest>(bind(&Server::onDeleteQueue, this, 
            placeholders::_1, placeholders::_2, placeholders::_3));
        // 绑定队列
        _dispatcher.registerMessageCallback<MessageQueue::queueBindRequest>(bind(&Server::onQueueBind, this, 
            placeholders::_1, placeholders::_2, placeholders::_3));
        // 解绑队列
        _dispatcher.registerMessageCallback<MessageQueue::queueUnBindRequest>(bind(&Server::onQueueUnBind, this, 
            placeholders::_1, placeholders::_2, placeholders::_3));
        // 消息发布
        _dispatcher.registerMessageCallback<MessageQueue::basicPublishRequest>(bind(&Server::onBasicPublish, this, 
            placeholders::_1, placeholders::_2, placeholders::_3));
        // 消息确认
        _dispatcher.registerMessageCallback<MessageQueue::basicAckRequest>(bind(&Server::onBasicAck, this, 
            placeholders::_1, placeholders::_2, placeholders::_3));
        // 队列消息订阅
        _dispatcher.registerMessageCallback<MessageQueue::basicConsumeRequest>(bind(&Server::onBasicConsume, this, 
            placeholders::_1, placeholders::_2, placeholders::_3));
        // 取消消息订阅
        _dispatcher.registerMessageCallback<MessageQueue::basicCancelRequest>(bind(&Server::onBasicCancel, this, 
            placeholders::_1, placeholders::_2, placeholders::_3));
        // 有消息到来时回调
        _server.setMessageCallback(bind(&ProtobufCodec::onMessage, _codec.get(),
            placeholders::_1, placeholders::_2, placeholders::_3));
        // 有连接到来时回调
        _server.setConnectionCallback(bind(&Server::onConnection, this, placeholders::_1));
    }
    
    // 启动服务器
    void start() 
    {
        _server.start();
        _baseloop.loop();
    }

private:
    //打开信道
    void onOpenChannel(const muduo::net::TcpConnectionPtr& conn, const openChannelRequestPtr& message, muduo::Timestamp) 
    {
        Connection::ptr mconn = _connection_manager->getConnection(conn);
        if (mconn.get() == nullptr) {
            lg(Debug, "打开信道时，没有找到连接对应的Connection对象！");
            conn->shutdown();
            return;
        }
        return mconn->openChannel(message);
    }

    //关闭信道
    void onCloseChannel(const muduo::net::TcpConnectionPtr& conn, const closeChannelRequestPtr& message, muduo::Timestamp) 
    {
        Connection::ptr mconn = _connection_manager->getConnection(conn);
        if (mconn.get() == nullptr) 
        {
            lg(Debug, "关闭信道时，没有找到连接对应的Connection对象！");
            conn->shutdown();
            return;
        }
        return mconn->closeChannel(message);
    }

    //声明交换机
    void onDeclareExchange(const muduo::net::TcpConnectionPtr& conn, const declareExchangeRequestPtr& message, muduo::Timestamp) 
    {
        Connection::ptr mconn = _connection_manager->getConnection(conn);
        if (mconn.get() == nullptr) 
        {
            lg(Debug, "声明交换机时，没有找到连接对应的Connection对象！");
            conn->shutdown();
            return;
        }
        Channel::ptr cp = mconn->getChannel(message->cid());
        if (cp.get() == nullptr) 
        {
            lg(Debug, "声明交换机时，没有找到信道！");
            return;
        }
        return cp->declareExchange(message);
    }

    //删除交换机
    void onDeleteExchange(const muduo::net::TcpConnectionPtr& conn, const deleteExchangeRequestPtr& message, muduo::Timestamp) 
    {
        Connection::ptr mconn = _connection_manager->getConnection(conn);
        if (mconn.get() == nullptr) 
        {
            lg(Debug, "删除交换机时，没有找到连接对应的Connection对象！");
            conn->shutdown();
            return;
        }
        Channel::ptr cp = mconn->getChannel(message->cid());
        if (cp.get() == nullptr) 
        {
            lg(Debug, "删除交换机时，没有找到信道！");
            return;
        }
        return cp->deleteExchange(message);
    }

    //声明队列
    void onDeclareQueue(const muduo::net::TcpConnectionPtr& conn, const declareQueueRequestPtr& message, muduo::Timestamp) 
    {
        Connection::ptr mconn = _connection_manager->getConnection(conn);
        if (mconn.get() == nullptr) 
        {
            lg(Debug, "声明队列时，没有找到连接对应的Connection对象！");
            conn->shutdown();
            return;
        }
        Channel::ptr cp = mconn->getChannel(message->cid());
        if (cp.get() == nullptr) 
        {
            lg(Debug, "声明队列时，没有找到信道！");
            return;
        }
        return cp->declareQueue(message);
    }

    //删除队列
    void onDeleteQueue(const muduo::net::TcpConnectionPtr& conn, const deleteQueueRequestPtr& message, muduo::Timestamp) 
    {
        Connection::ptr mconn = _connection_manager->getConnection(conn);
        if (mconn.get() == nullptr) 
        {
            lg(Debug, "删除队列时，没有找到连接对应的Connection对象！");
            conn->shutdown();
            return;
        }
        Channel::ptr cp = mconn->getChannel(message->cid());
        if (cp.get() == nullptr) 
        {
            lg(Debug, "删除队列时，没有找到信道！");
            return;
        }
        return cp->deleteQueue(message);
    }

    //队列绑定
    void onQueueBind(const muduo::net::TcpConnectionPtr& conn, const queueBindRequestPtr& message, muduo::Timestamp) 
    {
        Connection::ptr mconn = _connection_manager->getConnection(conn);
        if (mconn.get() == nullptr) 
        {
            lg(Debug, "队列绑定时，没有找到连接对应的Connection对象！");
            conn->shutdown();
            return;
        }
        Channel::ptr cp = mconn->getChannel(message->cid());
        if (cp.get() == nullptr) 
        {
            lg(Debug, "队列绑定时，没有找到信道！");
            return;
        }
        return cp->queueBind(message);
    }

    //队列解绑
    void onQueueUnBind(const muduo::net::TcpConnectionPtr& conn, const queueUnBindRequestPtr& message, muduo::Timestamp) 
    {
        Connection::ptr mconn = _connection_manager->getConnection(conn);
        if (mconn.get() == nullptr) 
        {
            lg(Debug, "队列解除绑定时，没有找到连接对应的Connection对象！");
            conn->shutdown();
            return;
        }
        Channel::ptr cp = mconn->getChannel(message->cid());
        if (cp.get() == nullptr) 
        {
            lg(Debug, "队列解除绑定时，没有找到信道！");
            return;
        }
        return cp->queueUnBind(message);
    }

    //消息发布
    void onBasicPublish(const muduo::net::TcpConnectionPtr& conn, const basicPublishRequestPtr& message, muduo::Timestamp) 
    {
        Connection::ptr mconn = _connection_manager->getConnection(conn);
        if (mconn.get() == nullptr) 
        {
            lg(Debug, "发布消息时，没有找到连接对应的Connection对象！");
            conn->shutdown();
            return;
        }
        Channel::ptr cp = mconn->getChannel(message->cid());
        if (cp.get() == nullptr) 
        {
            lg(Debug, "发布消息时，没有找到信道！");
            return;
        }
        return cp->basicPublish(message);
    }

    //消息确认
    void onBasicAck(const muduo::net::TcpConnectionPtr& conn, const basicAckRequestPtr& message, muduo::Timestamp) 
    {
        Connection::ptr mconn = _connection_manager->getConnection(conn);
        if (mconn.get() == nullptr) 
        {
            lg(Debug, "确认消息时，没有找到连接对应的Connection对象！");
            conn->shutdown();
            return;
        }
        Channel::ptr cp = mconn->getChannel(message->cid());
        if (cp.get() == nullptr) 
        {
            lg(Debug, "确认消息时，没有找到信道！");
            return;
        }
        return cp->basicAck(message);
    }

    //队列消息订阅
    void onBasicConsume(const muduo::net::TcpConnectionPtr& conn, const basicConsumeRequestPtr& message, muduo::Timestamp) 
    {
        Connection::ptr mconn = _connection_manager->getConnection(conn);
        if (mconn.get() == nullptr) 
        {
            lg(Debug, "队列消息订阅时，没有找到连接对应的Connection对象！");
            conn->shutdown();
            return;
        }
        Channel::ptr cp = mconn->getChannel(message->cid());
        if (cp.get() == nullptr) 
        {
            lg(Debug, "队列消息订阅时，没有找到信道！");
            return;
        }
        return cp->basicConsume(message);
    }

    //队列消息取消订阅
    void onBasicCancel(const muduo::net::TcpConnectionPtr& conn, const basicCancelRequestPtr& message, muduo::Timestamp) 
    {
        Connection::ptr mconn = _connection_manager->getConnection(conn);
        if (mconn.get() == nullptr) 
        {
            lg(Debug, "队列消息取消订阅时，没有找到连接对应的Connection对象！");
            conn->shutdown();
            return;
        }
        Channel::ptr cp = mconn->getChannel(message->cid());
        if (cp.get() == nullptr) 
        {
            lg(Debug, "队列消息取消订阅时，没有找到信道！");
            return;
        }
        return cp->basicCancel(message);
    }

    void onUnknownMessage(const muduo::net::TcpConnectionPtr& conn, const MessagePtr& message, muduo::Timestamp) 
    {
        LOG_INFO << "onUnknownMessage: " << message->GetTypeName();
        conn->shutdown();
    }

    void onConnection(const muduo::net::TcpConnectionPtr &conn) 
    {
        if (conn->connected()) 
        {
            _connection_manager->newConnection(_virtual_host, _consumer_manager, _codec, conn, _threadpool);
        }
        else 
        {
            _connection_manager->delConnection(conn);
        }
    }

private:
    muduo::net::EventLoop _baseloop;
    muduo::net::TcpServer _server;//服务器对象
    ProtobufDispatcher _dispatcher;//请求分发器对象--要向其中注册请求处理函数
    ProtobufCodecPtr _codec;//protobuf协议处理器--针对收到的请求数据进行protobuf协议处理
    VirtualHost::ptr _virtual_host;
    ConsumerManager::ptr _consumer_manager;
    ConnectionManager::ptr _connection_manager;
    threadpool::ptr _threadpool;
};
#endif